﻿using AutoMapper;
using Microsoft.Extensions.Logging;
using Msung.Business.IService;
using Msung.Core;
using Msung.Core.Caching;
using Msung.Core.Extensions;
using Msung.Core.Kafka;
using Msung.Core.Kafka.Integration;
using Msung.Core.Snowflake;
using Msung.Repository.Entity;
using Newtonsoft.Json;
using DataTraceService.IService;
using DataTraceService.Model;
using Msung.Business.IService.MSUNG;
using Msung.Business.Model;
using Msung.Business.Model.COLLECTION;
using Msung.Business.Service;

namespace DataTraceService.Consumers
{
    public class FlConsumerListener : IKafkaConsumerListener
    {
        readonly IRedisOperationRepository _redisRepository;
        readonly IFlowcardService _flowService;
        private readonly ILogger<FlConsumerListener> _logger;

        public FlConsumerListener(
            ILogger<FlConsumerListener> logger,
            IRedisOperationRepository redisRepository,
            IFlowcardService flowService
            )
        {
            this._redisRepository = redisRepository;
            _logger = logger;
            _flowService = flowService;
        }
        public void ConsumeAsync(RecieveResult recieveResult)
        {
            _logger.logInfo($"FLConsumerListener recieve message:{recieveResult.Message} location:{recieveResult.Offset} partition:{recieveResult.Partition} key:{recieveResult.Key} ");
            try
            {
                var model = recieveResult.GetRecieveMessage<FL_MODEL>();
                if (model == null)
                {
                    return;
                }
                UPBasketTask(model);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, ex.Message);
            }
            recieveResult.Commit();
        }
        #region 上/下花篮逻辑
        /// <summary>
        /// 上料处理逻辑
        /// </summary>
        /// <param name="model"></param>
        /// <returns></returns>
        public void UPBasketTask(FL_MODEL model)
        {
            string[] data = model.Struid.Split(',');
            if (!data.Any())
            {
                _logger.LogDebug("空数据：" + JsonConvert.SerializeObject(model));
                return;
            }
            foreach (var item in data)
            {
                if (!string.IsNullOrWhiteSpace(item.Trim()))
                {
                    string rediskey = $"FL:{item.Trim()}";
                    _redisRepository.Set(rediskey, model, TimeSpan.FromHours(8));
                }
            }
        }
        #endregion
    }
}
